package com.example.mqtt;

import io.moquette.broker.Server;
import io.moquette.broker.config.MemoryConfig;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.*;
import io.netty.buffer.ByteBuf;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

public class MqttServer {

    public static void main(String[] args) {
        Server server = new Server();
        
        // 创建消息拦截器，用于监听消息事件
        List<InterceptHandler> handlers = Collections.singletonList(new ServerInterceptHandler());
        
        try {
            // 创建配置
            Properties properties = new Properties();
            // MQTT TCP端口
            properties.setProperty("port", "1883");
            // WebSocket端口
            properties.setProperty("websocket_port", "8083");
            properties.setProperty("host", "0.0.0.0");
            properties.setProperty("allow_anonymous", "true");
            properties.setProperty("authenticator.class", "io.moquette.broker.security.PermitAllAuthenticator");
            properties.setProperty("persistence_enabled", "false");
            
            MemoryConfig config = new MemoryConfig(properties);
            
            // 启动服务器
            server.startServer(config, handlers);
            
            System.out.println("MQTT 服务端已启动");
            System.out.println("TCP 端口: 1883");
            System.out.println("WebSocket 端口: 8083");
            System.out.println("按任意键停止服务...");
            
            // 等待用户输入后关闭服务器
            System.in.read();
            
            server.stopServer();
            System.out.println("MQTT 服务端已停止");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    // 消息事件拦截处理器
    private static class ServerInterceptHandler implements InterceptHandler {
        @Override
        public String getID() {
            return "ServerInterceptHandler";
        }

        @Override
        public Class<?>[] getInterceptedMessageTypes() {
            return new Class<?>[] {
                InterceptConnectMessage.class,
                InterceptDisconnectMessage.class,
                InterceptConnectionLostMessage.class,
                InterceptPublishMessage.class,
                InterceptSubscribeMessage.class,
                InterceptUnsubscribeMessage.class,
                InterceptAcknowledgedMessage.class
            };
        }

        @Override
        public void onConnect(InterceptConnectMessage msg) {
            System.out.println("客户端连接: " + msg.getClientID());
        }

        @Override
        public void onDisconnect(InterceptDisconnectMessage msg) {
            System.out.println("客户端断开连接: " + msg.getClientID());
        }

        @Override
        public void onConnectionLost(InterceptConnectionLostMessage msg) {
            System.out.println("客户端连接丢失: " + msg.getClientID());
        }

        @Override
        public void onPublish(InterceptPublishMessage msg) {
            ByteBuf payload = msg.getPayload();
            byte[] bytes = new byte[payload.readableBytes()];
            payload.getBytes(payload.readerIndex(), bytes);
            String content = new String(bytes, StandardCharsets.UTF_8);
            
            System.out.println("收到发布消息:");
            System.out.println("  主题: " + msg.getTopicName());
            System.out.println("  内容: " + content);
            System.out.println("  QoS: " + msg.getQos());
        }

        @Override
        public void onSubscribe(InterceptSubscribeMessage msg) {
            System.out.println("客户端订阅: " + msg.getClientID());
            System.out.println("  主题: " + msg.getTopicFilter());
        }

        @Override
        public void onUnsubscribe(InterceptUnsubscribeMessage msg) {
            System.out.println("客户端取消订阅: " + msg.getClientID());
            System.out.println("  主题: " + msg.getTopicFilter());
        }

        @Override
        public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {
            // 消息确认
        }
    }
} 